package com.sdy.mq.config;

import lombok.Getter;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MQProducer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;

@Configuration
@ComponentScan({"com.sdy.mq.service"})
public class RocketMqConfig {
    @Value("${rocket-mq.nameserver.addr}")
    @Getter
    private String nameserverAddr;

    @Value("${rocket-mq.topic}")
    @Getter
    private String topic;
    
    @Value("${rocket-mq.send-msg-timeout:3000}")
    @Getter
    private Integer sendMsgTimeout;
    
    @Value("${rocket-mq.producer.group}")
    private String producerGroup;


    @Value("${rocket-mq.consumer.group}")
    private String consumerGroup;
    @Bean
    public MQProducer mqProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
        producer.setSendMsgTimeout(sendMsgTimeout);
        producer.setNamesrvAddr(nameserverAddr);
        producer.start();
        producer.setRetryTimesWhenSendAsyncFailed(0);
        return producer;
    }

    public DefaultMQPushConsumer createConsumerGroup(String groupName, MessageListener messageListener, String subscribeTopic, String subscribeTag, Integer thread, Integer maxBatchSize) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(nameserverAddr);
        consumer.subscribe(subscribeTopic, subscribeTag);
        consumer.setMessageListener(messageListener);
        consumer.setConsumeMessageBatchMaxSize(maxBatchSize);
        consumer.setConsumeThreadMax(thread);
        consumer.setConsumeThreadMin(thread);
        consumer.start();
        return consumer;
    }

    public DefaultMQPushConsumer createConsumerGroup(String groupName, MessageListener messageListener, String subscribeTopic, Integer thread, Integer maxBatchSize) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(nameserverAddr);
        consumer.subscribe(subscribeTopic, "*");
        consumer.setMessageListener(messageListener);
        consumer.setConsumeMessageBatchMaxSize(maxBatchSize);
        consumer.setConsumeThreadMax(thread);
        consumer.setConsumeThreadMin(thread);
        consumer.start();
        return consumer;
    }

    public DefaultMQPushConsumer createConsumerGroup(String groupName, MessageListener messageListener, String subscribeTopic, String subscribeTag) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(nameserverAddr);
        consumer.subscribe(subscribeTopic, subscribeTag);
        consumer.setMessageListener(messageListener);
        consumer.setConsumeMessageBatchMaxSize(1);
        consumer.start();
        return consumer;
    }

    public DefaultMQPushConsumer createConsumerGroup(String groupName, MessageListener messageListener, String subscribeTopic, Integer thread) throws Exception {
        return createConsumerGroup(groupName, messageListener, subscribeTopic, thread, 1);
    }

    public DefaultMQPushConsumer createConsumerGroup(String groupName, MessageListener messageListener, String subscribeTopic) throws Exception {
        return createConsumerGroup(groupName, messageListener, subscribeTopic, "*");
    }

    public DefaultMQPushConsumer createConsumer(MessageListener messageListener, String subscribeTopic) throws Exception {
        return createConsumerGroup(consumerGroup, messageListener, subscribeTopic, "*");
    }

    public DefaultMQPushConsumer createConsumer(MessageListener messageListener, String subscribeTopic, String subscribeTag) throws Exception {
        return createConsumerGroup(consumerGroup, messageListener, subscribeTopic, subscribeTag);
    }
}
